package com.ferry.transport.netty.handler.server;

import com.ferry.protocol.StatusCode;
import com.ferry.protocol.payload.RequestPayload;
import com.ferry.transport.netty.provider.ProviderProcessor;
import io.netty.channel.*;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by daiyong
 */
@ChannelHandler.Sharable
public class AcceptorHandler extends ChannelInboundHandlerAdapter {

	private static final Logger logger = LoggerFactory.getLogger(AcceptorHandler.class);

	private static final AtomicInteger channelCounter = new AtomicInteger(0);

	private ProviderProcessor providerProcessor;

	public AcceptorHandler(ProviderProcessor providerProcessor) {
		this.providerProcessor = providerProcessor;
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		Channel ch = ctx.channel();

		if (msg instanceof RequestPayload) {
			try {
				providerProcessor.handleRequest(ch, (RequestPayload) msg);
			} catch (Throwable t) {
				logger.error("handle msg error, channle", ch, t);
				providerProcessor.handleException(ch, (RequestPayload) msg, StatusCode.SERVER_ERROR, t);
			}
		} else {
			//释放消息
			ReferenceCountUtil.release(msg);
		}
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		channelCounter.incrementAndGet();
		super.channelActive(ctx);
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		channelCounter.decrementAndGet();
		super.channelInactive(ctx);
	}

	@Override
	public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
		Channel ch = ctx.channel();
		ChannelConfig config = ch.config();

		// 高水位线: ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK
		// 低水位线: ChannelOption.WRITE_BUFFER_LOW_WATER_MARK
		if (!ch.isWritable()) {
			// 当前channel的缓冲区(OutboundBuffer)大小超过了WRITE_BUFFER_HIGH_WATER_MARK
			logger.warn("{} is not writable, high water mask: {}, the number of flushed entries that are not written yet: {}.",
						ch, config.getWriteBufferHighWaterMark(), ch.unsafe().outboundBuffer().size());

			config.setAutoRead(false);
		} else {
			// 曾经高于高水位线的OutboundBuffer现在已经低于WRITE_BUFFER_LOW_WATER_MARK了
			logger.warn("{} is writable(rehabilitate), low water mask: {}, the number of flushed entries that are not written yet: {}.",
						ch, config.getWriteBufferLowWaterMark(), ch.unsafe().outboundBuffer().size());

			config.setAutoRead(true);
		}
	}


}
